package com.bw.day0630;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.TrafficPageViewBean;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

public class DwsTrafficVcChArIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //TODO 2.读取三个主题的数据创建流
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String ujTopic = "dwd_traffic_user_jump_detail";
        String pageTopic = "dwd_traffic_page_log";
        String groupId = "vccharisnew_pageview_window";

        //uv
        DataStreamSource<String> uvDs = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(uvTopic, groupId));
        DataStreamSource<String> ujDs = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(ujTopic, groupId));
        DataStreamSource<String> pageDs = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId));

        //TODO 3.统一数据格式
        //独立
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewUvDs = uvDs.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");

            return new TrafficPageViewBean(null, null,
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1l, 0l, 0l, 0l, 0l,
                    jsonObject.getLong("ts")
            );
        });

        //跳出
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewUjDs = ujDs.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");

            return new TrafficPageViewBean(null, null,
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0l, 0l, 0l, 0l, 1l,
                    jsonObject.getLong("ts")
            );
        });

        // 会话数 last_page_id=null 1、页面浏览数 1  和  浏览总时长 during_time

        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewPageDs = pageDs.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            JSONObject page = jsonObject.getJSONObject("page");

            String lastPageId = page.getString("last_page_id");
            long sv = 0l;
            if (lastPageId == null) {
                sv = 1l;
            }

            return new TrafficPageViewBean(null, null,
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0l,
                    sv,
                    1l,
                    page.getLong("during_time"),
                    0l,
                    jsonObject.getLong("ts")
            );

        });

        trafficPageViewUjDs.print("用户跳出数");
        trafficPageViewUvDs.print("独立访客数");
        trafficPageViewPageDs.print("page");

        //TODO 4.将三个流进行Union

        DataStream<TrafficPageViewBean> unionDs = trafficPageViewUvDs.union(
                trafficPageViewUjDs,
                trafficPageViewPageDs
        );

        //TODO 5.提取事件时间生成WaterMark

        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWmDs = unionDs.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                        return trafficPageViewBean.getTs();
                    }
                }));

        trafficPageViewWmDs.print("水位线");

        //TODO 6.分组开窗聚合

        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowedStream = trafficPageViewWmDs.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                return new Tuple4<>(
                        trafficPageViewBean.getAr(),
                        trafficPageViewBean.getCh(),
                        trafficPageViewBean.getIsNew(),
                        trafficPageViewBean.getVc()
                );
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10))).allowedLateness(Time.seconds(101));


        SingleOutputStreamOperator<TrafficPageViewBean> resDs = windowedStream.reduce(new ReduceFunction<TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean reduce(TrafficPageViewBean t1, TrafficPageViewBean t2) throws Exception {

                t1.setSvCt(t1.getSvCt() + t2.getSvCt());
                t1.setUvCt(t1.getUvCt() + t2.getUvCt());
                t1.setUjCt(t1.getUjCt() + t2.getUjCt());
                t1.setPvCt(t1.getPvCt() + t2.getPvCt());
                t1.setDurSum(t1.getDurSum() + t2.getDurSum());

                return t1;
            }
        }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow timeWindow, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                //获取数据
                TrafficPageViewBean next = iterable.iterator().next();
                //补充信息
                next.setStt(new Timestamp(timeWindow.getStart()));
                next.setEdt(new Timestamp(timeWindow.getEnd()));
                //修改TS
                next.setTs(System.currentTimeMillis());
                //输出数据
                collector.collect(next);
            }
        });



        //TODO 7.将数据写出到ClickHouse

        resDs.print(">>>>>.");
        resDs.addSink(MyClickHouseUtil.getSinkFunction("insert into  dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));





        env.execute();
    }
}